package com.alibaba.rocketmq.common.utils;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

import com.alibaba.rocketmq.common.ThreadFactoryImpl;
import org.apache.curator.ensemble.EnsembleProvider;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.CuratorFrameworkFactory.Builder;
import org.apache.curator.retry.RetryNTimes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;

/**
 *
 */
public class ZKClient{

	private static final Logger log = LoggerFactory.getLogger(ZKClient.class);

	private CuratorFramework m_client;

	private ZKConfig m_config;

	private void startCuratorFramework(){
		Builder builder = CuratorFrameworkFactory.builder();

		builder.connectionTimeoutMs(m_config.getZkConnectionTimeoutMillis());
		builder.maxCloseWaitMs(m_config.getZkCloseWaitMillis());
		builder.namespace(m_config.getZkNamespace());
		builder.retryPolicy(new RetryNTimes(m_config.getZkRetries(), m_config.getSleepMsBetweenRetries()));
		builder.sessionTimeoutMs(m_config.getZkSessionTimeoutMillis());
		builder.threadFactory(new ThreadFactoryImpl("CommonService-Zk"));
		builder.ensembleProvider(new EnsembleProvider() {

			@Override
			public void start() throws Exception {
			}

			@Override
			public String getConnectionString() {
				return m_config.getZkConnectionString();
			}

			@Override
			public void close() throws IOException {
			}

			@Override
			public void setConnectionString(String connectionString) {

			}

			@Override
			public boolean updateServerListEnabled() {
				return false;
			}
		});

		m_client = builder.build();
		m_client.start();
		try {
			m_client.blockUntilConnected();
			log.info("Conneted to zookeeper({}).", m_config.getZkConnectionString());
		} catch (InterruptedException e) {

 		}
	}

	public CuratorFramework get() {
		return m_client;
	}

}
